/*
   Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
   Original sources are available at www.latestbit.com

   Licensed under the Apache License, Version 2.0 (the "License");
   you may not use this file except in compliance with the License.
   You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

   Unless required by applicable law or agreed to in writing, software
   distributed under the License is distributed on an "AS IS" BASIS,
   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   See the License for the specific language governing permissions and
   limitations under the License.
*/

#ifndef THREADPOOL_HPP
#define THREADPOOL_HPP

#include <map>
#include <queue>

#include "cpputils/CppUtils.hpp"


namespace cpputils {
	namespace ept {

		/**
		 * The simple thread pool concept implementation.
		 * @par Usage:
		 * @code
		 * // Creation a thread pool.
		 * // Maximum count of threads in the pool is 10 and initially will be created is 3
		 * ThreadPool* pool = new ThreadPool ( 10 , 3 );
		 * // Creation task id
		 * ThreadPool::TaskId myTaskId = pool->createTaskId();
		 * // Scheduling the user task function in user class (boost::bind usage)
		 * // All scheduled tasks per task id can't run simultaneously
		 * pool->schedule (
		 *		myTaskId,
		 *      boost::bind ( &MyClass::myTaskHandler, this)
		 * );
		 * // Canceling all scheduled jobs by id
		 * pool->cancel ( myTaskId );
		 * // Wait for finish scheduled jobs by id
		 * pool->wait ( myTaskId );
		 * @endcode
		 * The implementation is thread-safe.
		 */
		class ThreadPool : public virtual RefObject {
		public:
			typedef boost::shared_ptr < ThreadPool > SharedPtr;
			typedef boost::thread Thread;
			typedef boost::function0 < void > Handler;
			typedef unsigned long TaskId;

		public:
			/**
			 * Creation a thread pool
			 * @param nInitialThreadCount Initial thread counts which creating immediately
			 * @param nMaxThreadCount Max thread count
			 */
			CPPU_DLLPRFX ThreadPool ( unsigned int nInitialThreadCount, unsigned int nMaxThreadCount, const std::string& sThreadPoolName="" );
			CPPU_DLLPRFX ~ThreadPool();

			/**
			 * Creation a task identifier for usage in other methods (schedule, cancel)
			 * Must be destroyed if more no needed.
			 */
			CPPU_DLLPRFX TaskId createTaskId();

			/**
			 * Destroy a task identifier
			 */
			CPPU_DLLPRFX void destroyTaskId(TaskId nTaskId);

			/**
			 * Scheduling a task without identifier.
			 * For all these type tasks the implementation have the system task identifiers
			 * and can be runned simultaneously.
			 */
			CPPU_DLLPRFX void schedule(Handler const & pfHandler);

			/**
			 * Scheduling a task (without user identifier) with N threads.
			 * For all these type tasks the implementation have the system task identifiers
			 * and can be runned simultaneously.
			 */
			CPPU_DLLPRFX void scheduleN(unsigned int nThreadCnt, Handler const & pfHandler);

			/**
			 * Scheduling a task for specified task id.
			 * @note All tasks per task id are not running simultaneously.
			 */
			CPPU_DLLPRFX void schedule(TaskId nTaskId, Handler const &pfHandler);

			/**
			 * Canceling tasks linked with task id
			 */
			CPPU_DLLPRFX void cancel(TaskId nTaskId);

			/**
			 * Cancel all tasks
			 */
			CPPU_DLLPRFX void cancelAll();

			/**
			 * Wait for finish all tasks for task id
			 */
			CPPU_DLLPRFX void wait(TaskId nTaskId);

			/**
			 * Wait for finish all tasks
			 */
			CPPU_DLLPRFX void waitAll();

			/**
			 * Wait for finish all tasks
			 */
			CPPU_DLLPRFX const std::string& getThreadPoolName();

		private:
			void createInitialThreads();
			void destroyMonitors();
			void checkThreadCounts();
			TaskId createSystemTaskId();
			void destroySystemTaskIds();

			void onRunTask(unsigned int m_nSystemTaskId);

		private:
			class TasksStorageItem {
			public:
				inline TasksStorageItem(TaskId nTaskId, const Handler& pfHandler):m_nTaskId(nTaskId),m_pfHandler(pfHandler) {};
				inline TasksStorageItem(const TasksStorageItem& src) { copyFrom(src); }
				inline void operator = (const TasksStorageItem& src) { copyFrom(src); }

				inline TaskId const& getTaskId() const { return m_nTaskId; }
				inline bool equalToTaskId (TaskId nTaskId) { return getTaskId() == nTaskId; }
				inline const Handler& getHandler() const { return m_pfHandler; }
				inline void executeHandler() { m_pfHandler(); };
			private:
				inline void copyFrom(const TasksStorageItem& src) { m_nTaskId = src.getTaskId(); m_pfHandler = src.m_pfHandler; }
			private:
				TaskId m_nTaskId;
				Handler m_pfHandler;
			};

			class TaskMonitor {
			public:
				inline TaskMonitor(): m_poParent(NULL), m_nTaskCount(0), m_mxMonitor(NULL), m_bRunningStatus(false), m_bDestroyOnFinish(false) {}
				inline TaskMonitor(ThreadPool* poParent):m_poParent ( poParent ), m_nTaskCount(0),m_mxMonitor(new boost::recursive_mutex()), m_bRunningStatus(false), m_bDestroyOnFinish(false) {};
				inline TaskMonitor(const TaskMonitor& src):m_poParent(NULL), m_nTaskCount(0), m_mxMonitor(NULL), m_bRunningStatus(false), m_bDestroyOnFinish(false) { copyFrom(src); }

				inline void operator = (const TaskMonitor& src) { copyFrom(src); }

				inline unsigned int getTaskCount() const { boost::mutex::scoped_lock oLock(m_poParent->m_mxTaskMonitorCounters); return m_nTaskCount; }
				inline void setTaskCount(unsigned int value) { boost::mutex::scoped_lock oLock(m_poParent->m_mxTaskMonitorCounters); m_nTaskCount = value; }
				inline unsigned int incTaskCount() { boost::mutex::scoped_lock oLock(m_poParent->m_mxTaskMonitorCounters); return m_nTaskCount++; }
				inline unsigned int decTaskCount() { boost::mutex::scoped_lock oLock(m_poParent->m_mxTaskMonitorCounters); return m_nTaskCount--; }

				inline boost::recursive_mutex& getMonitor() const { return *m_mxMonitor;}
				inline void destroy() { m_nTaskCount = 0; delete m_mxMonitor; m_mxMonitor=NULL; }
				inline void setRunning(bool bValue) { m_bRunningStatus = bValue; }
				inline bool isRunning() { return m_bRunningStatus; }
				inline void setDestroyOnFinish(bool bValue) { m_bDestroyOnFinish = bValue ; }
				inline bool isDestroyOnFinish() { return m_bDestroyOnFinish; }
			private:
				inline void copyFrom(const TaskMonitor& src) { m_poParent = src.m_poParent; m_nTaskCount = src.m_nTaskCount; m_mxMonitor = src.m_mxMonitor; m_bRunningStatus = src.m_bRunningStatus; m_bDestroyOnFinish = src.m_bDestroyOnFinish; }
			private:
				ThreadPool* m_poParent;
				unsigned int m_nTaskCount;
				boost::recursive_mutex* m_mxMonitor;
				bool m_bRunningStatus;
				bool m_bDestroyOnFinish;
			};

		private:
			unsigned int m_nInitialThreadCount;
			unsigned int m_nMaxThreadCount;
			unsigned int m_nCurrentThreadCount;

			typedef std::map< TaskId, TaskMonitor > TaskMonitors;
			TaskMonitors m_mTaskMonitors;
			boost::recursive_mutex m_mxTaskMonitors;
			boost::mutex m_mxTaskMonitorCounters;

			typedef std::list < TasksStorageItem > TaskStorage; // deque / map cannot be approach for erasing by id in the loop
			TaskStorage m_aTaskStorage;
			unsigned int m_nTaskStorageSize;
			TaskId m_nTaskIdCounter;
			boost::recursive_mutex m_mxTaskStorage;


			boost::condition m_evTaskAvailables;

			typedef std::vector<TaskId> SystemTaskIds;
			SystemTaskIds m_aSystemTaskIds;
			SystemTaskIds::iterator m_itCurrentTaskId;
			boost::recursive_mutex m_mxSystemTaskId;


			boost::thread_group* m_poThreadGroup;
			bool m_bStopFlag;
			const std::string m_sThreadPoolName;
		};

	}
}

#endif
